{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "K8S_PROXY_ADDR='127.0.0.1:8001'\n",
    "K8S_NAMESPACE='mdt'\n",
    "APP_NAME='pyspark-classification'\n",
    "MODEL_NAME='clipper-pyspark-predict'\n",
    "REPO_URL='658391232643.dkr.ecr.us-west-2.amazonaws.com'\n",
    "VERSION = 1\n",
    "\n",
    "SPARK_URL='ae092775dc67111e8aee806444a30499-1743429263.us-west-2.elb.amazonaws.com'\n",
    "SPARK_APP='Clipper'\n",
    "HADOOP_URL='ad95fe885c37011e8aee806444a30499-1181034928.us-west-2.elb.amazonaws.com'\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Connect to clipper and register App"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from clipper_admin import ClipperConnection, KubernetesContainerManager\n",
    "from clipper_admin.deployers import python as python_deployer\n",
    "manager = KubernetesContainerManager(\n",
    "    kubernetes_proxy_addr=K8S_PROXY_ADDR, namespace=K8S_NAMESPACE)\n",
    "clipper_conn = ClipperConnection(manager)\n",
    "clipper_conn.connect()\n",
    "\n",
    "clipper_conn.register_application(\n",
    "   name = APP_NAME, input_type = 'doubles', default_output = '0', slo_micros = 100000000)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create a prediction function and deploy it to clipper"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark\n",
    "from clipper_admin.deployers.pyspark import deploy_pyspark_model\n",
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml import PipelineModel\n",
    "from pyspark.sql import SparkSession\n",
    "import numpy as np\n",
    "\n",
    "sparkUrl = \"spark://{}:7077\".format(SPARK_URL)\n",
    "conf = pyspark.SparkConf().setAppName(SPARK_APP).setMaster(sparkUrl)\n",
    "sc = pyspark.SparkContext(conf=conf)\n",
    "\n",
    "spark = SparkSession.builder.master(SPARK_URL) \\\n",
    "    .appName(SPARK_APP) \\\n",
    "    .getOrCreate()\n",
    "\n",
    "# Loading a training dataset omitted...\n",
    "#pipeline = Pipeline.read().load(\"hdfs://{}:9000/tmp/classification-pipeline\".format(SPARK_URL))\n",
    "model = PipelineModel.read().load(\"hdfs://{}:9000/tmp/classification-model\".format(SPARK_URL))\n",
    "\n",
    "# Note that this function accesses the trained PySpark model via an explicit\n",
    "# argument, but other state can be captured via closure capture if necessary.\n",
    "def predict(spark, model, inputs):\n",
    "    return [str(model.transform(x)) for x in inputs]\n",
    "\n",
    "deploy_pyspark_model(\n",
    "    clipper_conn,\n",
    "    name=MODEL_NAME,\n",
    "    input_type=\"doubles\",\n",
    "    func=predict,\n",
    "    pyspark_model=model,\n",
    "    sc=sc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Link the model to application"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "clipper_conn.link_model_to_app(app_name=APP_NAME, model_name=MODEL_NAME)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import requests, json, numpy as np\n",
    "def predict():\n",
    "    headers = {\"Content-type\": \"application/json\"}\n",
    "    data=json.dumps({\"input\": [5.9, 3.0, 5.1, 1.8]})\n",
    "    print(data)\n",
    "    url = \"http://{}/api/v1/namespaces/{}/services/query-frontend-at-default-cluster:1337/proxy/{}/predict\".format(K8S_PROXY_ADDR,K8S_NAMESPACE,APP_NAME)\n",
    "    res = requests.post(url, headers=headers, data=data)\n",
    "    return res.json()\n",
    "\n",
    "print(predict())"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
